# Copyright (c) 2018 Presto Labs Pte. Ltd.
# Author: xguo

import cachetools
import collections
import enum
import functools
import typing

import requests
import recordclass
import tornado.gen

import coin.flow.topic_map as topic_map
import coin.flow.subscriber as flow_subscriber
import coin.proto.coin_market_enums_pb2 as coin_enum

from coin.base.config import Config
from coin.base.timestamp import get_timestamp
from coin.exchange.base.order_fill_util import FillManager
from coin.exchange.base.order_gateway_ready import PositionOrderGatewayReady
from coin.exchange.base.rest.task_scheduler import Scheduler, wrap_async_task
from coin.exchange.base.rest_order_gateway_base import PrivateExchangeMessage
from coin.exchange.huobi.kr_rest.currency import HuobiCurrency
from coin.exchange.huobi_futures.kr_rest.futures_product import HuobiFuturesProduct
from coin.exchange.huobi_futures.kr_rest.private_client import HuobiFuturesPrivateParser
from coin.exchange.huobi_futures.ws.ws_private_client import HuobiFuturesWsPrivateClient
from coin.exchange.util.rest_rate_limit import RestRateLimiter
from coin.proto.coin_order_gateway_pb2 import OrderGatewayConfig, OrderEvent
from coin.strategy.hk_hard_arb.og_util import print_working_orders

from coin.proto.coin_query_pb2 import (
    AccountBalance,
    AccountPosition,
    ProductPosition,
)
from coin.exchange.huobi_futures.kr_rest.native_private_client import (
    HuobiFuturesDirectionType,
    HuobiFuturesNativePrivateClient,
    HuobiFuturesOffsetType,
    HuobiFuturesOrderPriceType,
    HuobiFuturesOrderSide,
    HuobiFuturesStatusType,
)
from coin.exchange.base.order_gateway import (
    BaseOrderGatewayInfoMutator,
    Order,
    OrderGatewayStatus,
    OrderType,
)
from coin.exchange.base.rest_order_gateway_base import (
    RestOrderGateway,
    create_order_internal,
    exchange_message_handler,
    coin_order_pb2,
)


def translate_order_side(hbdm_order_side: HuobiFuturesOrderSide):
  direction, offset = hbdm_order_side.name.lower().split('_')
  return HuobiFuturesDirectionType(direction), HuobiFuturesOffsetType(offset)


def _parse_order_property(message):
  if message['order_price_type'] == 1:
    order_type = HuobiFuturesOrderPriceType.LIMIT
  else:
    order_type = HuobiFuturesOrderPriceType(message['order_price_type'])
  direction = message['direction'].upper()
  offset = message['offset'].upper()
  order_side = getattr(HuobiFuturesOrderSide, '%s_%s' % (direction, offset))
  if order_type in (HuobiFuturesOrderPriceType.LIMIT, HuobiFuturesOrderPriceType.POST_ONLY):
    order_type = OrderType.LIMIT
  return {"order_type": order_type, "order_side": order_side}


def create_order_from_open_order_message(message, timestamp, order_id=None):
  """
  {
  "status": "ok",
  "data": {
    "orders": [
      {
        "client_order_id": null,
        "contract_code": "BTC190329",
        "contract_type": "quarter",
        "created_at": 1544785876968,
        "direction": "buy",
        "fee": 0,
        "lever_rate": 20,
        "margin_frozen": 0.0025,
        "offset": "open",
        "order_id": 14,
        "order_price_type": "limit",
        "order_source": "web"
        "order_type": 1,
        "price": 2000,
        "profit": 0,
        "status": 3,
        "symbol": "BTC",
        "trade_avg_price": null,
        "trade_turnover": 0,
        "trade_volume": 0,
        "volume": 1,
      }
    ],
    "total_page": 1,
    "current_page": 1,
    "total_size": 1
  },
  "ts": 1544786335740
}
  """
  price_orig = float(message['price'])
  qty_orig = float(message['volume'])
  product = HuobiFuturesProduct.FromStrNativeProduct(message['contract_code'])
  exchange_order_id = convert_to_exchange_order_id(product, message['order_id'])
  qty_accum_fill = float(message['trade_volume'])
  if message['trade_avg_price']:
    price_avg_fill = float(message['trade_avg_price'])
  else:
    price_avg_fill = None
  fee = float(message['fee'])
  parsed_result = _parse_order_property(message)
  status = HuobiFuturesStatusType(message['status'])

  fill = FillManager(
      price_orig=price_orig,
      qty_orig=qty_orig,
      price_last_fill=price_avg_fill,
      qty_last_fill=qty_accum_fill,
      price_avg_fill=price_avg_fill,
      qty_accum_fill=qty_accum_fill,
      timestamp_last_fill=timestamp,
  )

  order = Order(order_id=order_id,
                product=product,
                price=price_orig,
                qty=qty_orig,
                order_side=parsed_result['order_side'],
                order_type=parsed_result['order_type'],
                tag=None,
                accepted=True,
                cancel_sent=False,
                internal=create_order_internal(
                    timestamp=timestamp,
                    exchange_order_id=exchange_order_id,
                    fill=fill,
                    fee=fee,
                    status=status,
                ))
  return order


# `real_exchange_order_id` refers to the order_id chosen by the exchange.
# It is a very simple number, like 14, and looks like unique for one particular
# product and user account. Since we assume exchange order id is a unique id
# for all orders of one account. So we need to convert
# real_exchange_order_id to exchange_order_id.
def get_real_exchange_order_id(order):
  if order.internal.exchange_order_id is not None:
    return order.internal.exchange_order_id.split('_')[1]
  else:
    return None


def convert_to_exchange_order_id(product, real_exchange_order_id):
  return '%s_%s' % (product.symbol, real_exchange_order_id)


FuturesPosition = recordclass.recordclass('FuturesPosition',
                                          ['long', 'short', 'long_avg_price', 'short_avg_price'])


def create_futures_positions(long=0., short=0., long_avg_price=None, short_avg_price=None):
  return FuturesPosition(long=long,
                         short=short,
                         long_avg_price=long_avg_price,
                         short_avg_price=short_avg_price)


class HuobiFuturesOrderGatewayInfo(BaseOrderGatewayInfoMutator):
  CurrencyType = HuobiCurrency
  ProductType = HuobiFuturesProduct
  BalanceType = float
  PositionType = float

  def __init__(self,
               currencies: typing.List[HuobiCurrency],
               products: typing.List[HuobiFuturesProduct]):
    super().__init__(currencies, products)
    self._unrealized_pnl = collections.defaultdict(float)
    self._avg_entry_price = collections.defaultdict(create_futures_positions)

  def update_balance(self, message):
    msg = HuobiFuturesPrivateParser.parse_balance(message)
    for balance in msg.each_balance:
      currency = HuobiCurrency.FromStrNativeCurrency(balance.currency_native)
      if currency in self.currencies:
        self.set_balance(currency, balance.total, balance)

  def update_position(self, message):
    for product in self.products:
      position = ProductPosition(symbol=product.symbol,
                                 long_position=0,
                                 short_position=0,
                                 net_position=0)
      self.set_position(product, 0.0, position)

    msg = HuobiFuturesPrivateParser.parse_account_position(message)
    for position in msg.each_position:
      product = HuobiFuturesProduct.FromStrProduct(position.symbol)
      if product in self.products:
        self.set_position(product, position.net_position, position)

    self._unrealized_pnl = collections.defaultdict(float)
    self._avg_entry_price = collections.defaultdict(create_futures_positions)
    for msg in message['data']:
      currency = msg['symbol']
      direction = HuobiFuturesDirectionType(msg['direction'])
      product = HuobiFuturesProduct.FromStrNativeProduct(msg['contract_code'])
      self._unrealized_pnl[currency] += float(msg['profit_unreal'])
      if direction == HuobiFuturesDirectionType.BUY:
        self._avg_entry_price[product].long = float(msg['volume'])
        self._avg_entry_price[product].long_avg_price = float(msg['cost_open'])
      elif direction == HuobiFuturesDirectionType.SELL:
        self._avg_entry_price[product].short = float(msg['volume'])
        self._avg_entry_price[product].short_avg_price = float(msg['cost_open'])

  def balance_as_proto(self):
    return AccountBalance(exchange='Huobi', each_balance=self.get_notNone_balances_internal())

  def position_as_proto(self):
    return AccountPosition(exchange='Huobi', each_position=self.get_notNone_positions_internal())

  def get_product_position_proto(self, product):
    return self._positions_internal[product.symbol]

  def get_futures_position(self, product):
    """return FuturesPosition namedtuple"""
    return self._avg_entry_price.get(product, None)

  # Margin balance includes unrealized pnl, but this is provided by exchange
  # and could be out of date or slow.
  # DO NOT USE THIS IN TRADING, ONLY ACCEPTABLE IN PNL DISPLAY.
  def get_margin_balance(self, currency: str):
    return self._unrealized_pnl[currency] + self.get_balance(currency)


class HuobiFuturesOrderGateway(RestOrderGateway):
  _exchange = 'Huobi'
  _market_type = 'Futures'
  _api_version = None

  OGInfoType = HuobiFuturesOrderGatewayInfo

  class _MsgType(enum.IntEnum):
    REST_ACCOUNT_BALANCE = 0
    REST_ACCOUNT_POSITION = 1
    OPEN_ORDERS = 2
    ALL_ORDERS = 3
    ORDERS_INFO = 4
    WS_ORDER_UPDATE = 5

  def __init__(self,
               currencies: typing.List[HuobiCurrency],
               products: typing.List[HuobiFuturesProduct],
               config: Config,
               og_config=None,
               logger=None):
    super().__init__(currencies, products, config, og_config, logger)
    self._og_ready = PositionOrderGatewayReady()
    self._og_ready.set_status(OrderGatewayStatus.INITIALIZING)

    # Following instance attributes are initialized in start method.
    self._flow_sub = None
    self._ioloop = None
    self._publisher = None
    self._scheduler = None
    self._private_client = None
    self._ws_private_client = None
    self._rest_rate_limiter = RestRateLimiter(delay=10)
    self._use_order_info = True

    self._og_config = OrderGatewayConfig(
        account_query_period=10,
        user_query_period=1,
        orders_snapshot_query_period=10,
        fill_query_period=2,
        rest_lib='pycurl',
    )
    if og_config is not None:
      self._og_config.MergeFrom(og_config)
    self._og_logger.exchange_type = coin_enum.Huobi
    self._recent_orders = collections.defaultdict(lambda: collections.deque(maxlen=50))
    self._buy_close_pos_insufficient_pos_error = cachetools.TTLCache(maxsize=2, ttl=4)
    self._sell_close_pos_insufficient_pos_error = cachetools.TTLCache(maxsize=2, ttl=4)
    self.recent_cancel_accepted = cachetools.TTLCache(maxsize=10, ttl=4)

  def submittable(self):
    return self.is_ready() and self._rest_rate_limiter.is_available()

  def has_recent_close_error(self, side):
    if side == HuobiFuturesOrderSide.BUY_CLOSE:
      return len(self._buy_close_pos_insufficient_pos_error) > 0
    elif side == HuobiFuturesOrderSide.SELL_CLOSE:
      return len(self._sell_close_pos_insufficient_pos_error) > 0
    else:
      raise RuntimeError('Unknown side %s', side)

  def get_huobi_futures_position(self, product: HuobiFuturesProduct):
    return self._og_info.get_futures_position(product)

  def start(self, queue, ioloop, flow_sub=None):
    api_host = self._og_config.rest_root_url if \
               self._og_config.HasField('rest_root_url') else None
    disable_ssl = self._og_config.rest_disable_ssl

    topic_id = topic_map.search_topic_id('huobi_futures_rest_private_1')[0][1]
    self._ioloop = ioloop

    self._publisher = queue.get_writer(topic_id=topic_id)
    self._scheduler = Scheduler(ioloop=ioloop, logger=self._logger)

    self._private_client = HuobiFuturesNativePrivateClient(key_file=self._config.key_file,
                                                           api_host=api_host,
                                                           disable_ssl=disable_ssl,
                                                           use_async=True,
                                                           rest_lib=self._og_config.rest_lib)
    self._prepare_ws_private_client(start=True)
    self._install_periodic_query()

    # Subscribe to flow for feed.
    self._flow_sub = flow_sub or flow_subscriber.from_queue(queue)
    self._flow_sub.subscribe('huobi_futures_rest_private_1', self.on_private_exchange_msg)

  @tornado.gen.coroutine
  def _install_periodic_query(self):
    assert self._scheduler is not None, 'self._scheduler is not set!'
    assert self._publisher is not None, 'self._publisher is not set!'

    query_balance_task = wrap_async_task(
        self._private_client.query_account_info,
        msg_type=self._MsgType.REST_ACCOUNT_BALANCE,
        publisher=self._publisher,
    )

    # query balance
    self._scheduler.add_periodic_task(
        period=self._og_config.account_query_period,
        job_func=query_balance_task,
    )

    yield tornado.gen.sleep(self._og_config.burst_query_avoid_time)
    query_position_task = wrap_async_task(
        self._private_client.query_position_info,
        msg_type=self._MsgType.REST_ACCOUNT_POSITION,
        publisher=self._publisher,
    )

    # query positions
    self._scheduler.add_periodic_task(
        period=self._og_config.user_query_period,
        job_func=query_position_task,
    )

    symbols = list(set([prod.base.native_currency_futures for prod in self._og_info.products]))
    # query open orders
    for symbol in symbols:
      query_orders_task = wrap_async_task(self._private_client.query_open_orders,
                                          msg_type=self._MsgType.OPEN_ORDERS,
                                          publisher=self._publisher,
                                          func_params=dict(symbol=symbol))
      self._scheduler.add_periodic_task(period=self._og_config.orders_snapshot_query_period,
                                        job_func=query_orders_task)

    yield tornado.gen.sleep(self._og_config.burst_query_avoid_time)
    for symbol in symbols:
      if self._use_order_info:
        query_orders_info_task = wrap_async_task(self.query_order_info,
                                                 msg_type=self._MsgType.ORDERS_INFO,
                                                 publisher=self._publisher,
                                                 func_params=dict(symbol=symbol))

        self._scheduler.add_periodic_task(
            period=self._og_config.fill_query_period,
            job_func=query_orders_info_task,
        )
      else:
        query_orders_task = wrap_async_task(self._private_client.query_historical_orders,
                                            msg_type=self._MsgType.ALL_ORDERS,
                                            publisher=self._publisher,
                                            func_params=dict(symbol=symbol,
                                                             trade_type=0,
                                                             order_status_type=1,
                                                             order_status=0,
                                                             create_date=7,
                                                             page_index=1,
                                                             page_size=50))

        self._scheduler.add_periodic_task(
            period=self._og_config.fill_query_period,
            job_func=query_orders_task,
        )

    def print_info():
      print_working_orders(self._logger, self)

    self._scheduler.add_periodic_task(
        period=10,
        job_func=print_info,
    )
    self._scheduler.start()

  def _prepare_ws_private_client(self, start=False):
    if self._ws_private_client is not None:
      self._ws_private_client.stop()

    api_host = self._og_config.rest_root_url if self._og_config.HasField('rest_root_url') else None
    disable_ssl = self._og_config.rest_disable_ssl
    self._ws_private_client = HuobiFuturesWsPrivateClient(
        self._config.key_file,
        ioloop=self._ioloop,
        notification_callback=self._notification_callback,
        close_callback=functools.partial(self._prepare_ws_private_client, True),
        on_ready=self._on_ws_private_client_ready,
        api_host=api_host,
        disable_ssl=disable_ssl,
    )

    if start:
      self._ws_private_client.start()

  def _on_ws_private_client_ready(self):
    native_symbols = list(set([prod.base.native_currency for prod in self._og_info.products]))
    self._ws_private_client.prepare_subscription(native_symbols)

  def _notification_callback(self, data):
    assert data['op'] == 'notify', data
    timestamp = get_timestamp()
    topic = data['topic']
    if topic.startswith('orders.'):
      msg_type = self._MsgType.WS_ORDER_UPDATE
    else:
      self._logger.error('Unknown subscription topic! %s', data)
      return

    exchange_message = PrivateExchangeMessage(msg_type=msg_type,
                                              data=data,
                                              received_timestamp=timestamp,
                                              request_params=None,
                                              request_timestamp=None)
    self._publisher.write(timestamp, exchange_message)

  def query_order_info(self, symbol):
    order_id_list = [
        get_real_exchange_order_id(order) for order in self._recent_orders[symbol] if order.accepted
    ]
    if len(order_id_list) == 0:
      return None

    assert len(order_id_list) <= 50
    return self._private_client.query_order_info(symbol=symbol, order_id_list=order_id_list)

  def _async_submit_order(self, order, post_only):
    assert order.order_id is not None, order
    assert order.order_type == OrderType.LIMIT, order

    order_price_type = HuobiFuturesOrderPriceType.LIMIT
    if post_only:
      order_price_type = HuobiFuturesOrderPriceType.POST_ONLY

    formatter = self._order_formatter[order.product]
    direction, offset = translate_order_side(order.order_side)
    params = {
        'contract_code': order.product.native_symbol,
        'price': formatter.format_price(order.price),
        'volume': formatter.format_qty(order.qty),
        'direction': direction,
        'offset': offset,
        'leverage_rate': order.internal.leverage,
        'order_price_type': order_price_type,
    }
    fut = self._private_client.place_order(**params)
    key = order.product.base.native_currency_futures
    self._recent_orders[key].append(order)
    return fut

  def _async_cancel_order(self, order):
    assert order.internal.exchange_order_id is not None, order
    real_exchange_order_id = get_real_exchange_order_id(order)
    fut = self._private_client.cancel_order(
        symbol=order.product.base.native_currency_futures,
        order_id=real_exchange_order_id,
    )
    return fut

  def _handle_submit_response(self, order, response):
    """{'status': 'ok', 'data': {'order_id': 35}, 'ts': 1545046061904}"""
    if response.status_code != requests.codes.ok:
      self._order_manager.remove_order(order, ignore_error=True)
      if response.status_code == 429:
        self._rest_rate_limiter.set_error(response.status_code)
        self._logger.error('Rate limit hit!')
      else:
        self._logger.error('Submit order %s failed!\n%s', order, response.content)
      return OrderEvent.ORDER_REJECTED

    message = response.json()
    if message['status'] != 'ok':
      self._logger.error('Status is not ok! %s\n%s', message, order)
      if message['err_code'] == 1048:
        if order.order_side == HuobiFuturesOrderSide.BUY_CLOSE:
          self._buy_close_pos_insufficient_pos_error[order.order_id] = True
        elif order.order_side == HuobiFuturesOrderSide.BUY_CLOSE:
          self._sell_close_pos_insufficient_pos_error[order.order_id] = True
        else:
          self._logger.error('Unexpected order_side:\n %s', order)

      self._order_manager.remove_order(order, ignore_error=True)
      return OrderEvent.ORDER_ERROR

    exchange_order_id = convert_to_exchange_order_id(order.product, message['data']['order_id'])
    prev_order = self._order_manager.get(exchange_order_id=exchange_order_id)
    if prev_order is None:
      self._order_manager.set_order(order, exchange_order_id=exchange_order_id)
    else:
      if prev_order.order_id != order.order_id:
        self._logger.info('order id replaced: %s => %s', prev_order.order_id, order.order_id)
        self._order_manager.remove_order(prev_order)
        self._order_manager.set_order(order, exchange_order_id=exchange_order_id)
        self._order_manager.remove_order(order)
        prev_order.order_id = order.order_id
        self._order_manager.add_order(prev_order)
    return OrderEvent.ORDER_ACCEPTED

  def _handle_cancel_response(self, order, response):
    """
    {
    "status": "ok",
    "data": {
      "errors": [],
      "successes": "27"
    },
    "ts": 1545045423009
    }
    """
    if response.status_code != requests.codes.ok:
      self._logger.error('Cancel order failed with http response code=%s!. %s' %
                         (response.status_code, response.content))
      return OrderEvent.CANCEL_ERROR

    message = response.json()
    if message['status'] != 'ok':
      self._logger.error('Cancel order (%s) error! %s', order.order_id, message)
      if self._order_manager.is_dead_order(order):
        self._logger.warning('Order (%s) is dead already!', order.order_id)
      else:
        self._logger.error('Unexpected error! Need inspection! %s', message)
      return OrderEvent.CANCEL_ERROR

    data = message['data']
    errors = data.get('errors')  # Empty list if no error
    exchange_order_id = convert_to_exchange_order_id(order.product, data.get('successes'))
    if errors:
      if errors[0]['err_msg'] == 'The order has been executed.':
        return OrderEvent.CANCEL_ERROR
      else:
        self._logger.error('Cancel error: %s', message)
        return OrderEvent.CANCEL_ERROR
    elif exchange_order_id != order.internal.exchange_order_id:
      self._logger.error('Insane response, exchange order id not consistent: %s, %s',
                         order,
                         message)
      return OrderEvent.CANCEL_ERROR
    else:
      if self._order_manager.is_dead_order(order):
        self._logger.info('Order is dead already. %s', order.order_id)
    self.recent_cancel_accepted[order.order_id] = order
    return OrderEvent.CANCEL_ACCEPTED

  def _update_order(self, order, order_info, timestamp, fill_type=None):
    status = order_info.internal.status
    exchange_order_id = order_info.internal.exchange_order_id
    if order.accepted is False:
      self._order_manager.set_order(order, exchange_order_id=exchange_order_id)

    if status == HuobiFuturesStatusType.NEW:
      pass
    elif status == HuobiFuturesStatusType.PARTIAL_FILLED:
      pass
    elif status == HuobiFuturesStatusType.FULLY_FILLED:
      if not self._order_manager.is_dead_order(order):
        self._order_manager.set_order(order, fully_filled=True)
    elif status in (HuobiFuturesStatusType.PARTIAL_CANCELED, HuobiFuturesStatusType.CANCELED):
      if not self._order_manager.is_dead_order(order):
        self._og_logger.gen_order_event_and_log(order, timestamp, OrderEvent.CANCEL_CONFIRMED)
        self._order_manager.set_order(order, cancel_confirmed=True)
    else:
      self._logger.info('%s(%s) status: %s',
                        order.order_id,
                        order.internal.exchange_order_id,
                        status)

    price_avg_fill = order_info.internal.fill.value.price_avg_fill
    qty_accum_fill = order_info.internal.fill.value.qty_accum_fill
    last_fill = order.internal.fill.update_by_accum_fill(price_avg_fill, qty_accum_fill)
    if last_fill:
      self._publish_fill(last_fill.price_last_fill, last_fill.qty_last_fill, order)
      self._og_logger.gen_order_filled_event_and_log(order, timestamp, fill_type=fill_type)
      order.internal.fill.log_last_fill(order, self._logger)
    elif order is order_info and order.internal.status in (HuobiFuturesStatusType.PARTIAL_FILLED,
                                                           HuobiFuturesStatusType.FULLY_FILLED):
      price_last_fill = order.internal.fill.value.price_last_fill
      qty_last_fill = order.internal.fill.value.qty_last_fill
      self._publish_fill(price_last_fill, qty_last_fill, order)
      self._og_logger.gen_order_filled_event_and_log(order, timestamp, fill_type=fill_type)
      order.internal.fill.log_last_fill(order, self._logger)

  def _handle_error_response(self, response):
    if response.status_code == 504:
      self._logger.error('504 Gateway Timeout error!')
    else:
      self._logger.error('response error! %s, %s, %s',
                         response.status_code,
                         response.content,
                         response.url)

  @exchange_message_handler(_MsgType.OPEN_ORDERS)
  def _handle_open_orders(self, exchange_message: PrivateExchangeMessage):
    """{'status': 'error', 'err_code': 1032, 'err_msg': 'The number of access exceeded the limit.', 'ts': 1545046221833}"""
    timestamp = get_timestamp()
    response = exchange_message.data
    if response.status_code != requests.codes.ok:
      self._handle_error_response(response)
      return

    message = response.json()
    if message['status'] != 'ok':
      self._logger.error('status is not ok: %s' % message)
      return

    assert message['data']['total_page'] == 1, message
    exchange_open_order_dict = {}
    products = set()
    for entry in message['data']['orders']:
      native_symbol = entry['contract_code']
      product = HuobiFuturesProduct.FromStrNativeProduct(native_symbol)
      if product not in self._og_info.products:
        continue  # ignore

      order = create_order_from_open_order_message(entry, timestamp)
      exchange_order_id = order.internal.exchange_order_id
      prev_order = self._order_manager.get(exchange_order_id=exchange_order_id)
      if prev_order is None:
        order_id = self._order_id_generator.gen_order_id(is_foreign=True)
        order.order_id = order_id
      else:
        order.order_id = prev_order.order_id
      exchange_open_order_dict[exchange_order_id] = order

    self._order_manager.copy_from_exchange_open_orders(exchange_open_order_dict, products)

    if not self.is_ready():
      self._og_ready.set_orders_ready(True)
      self._og_ready.check_ready()

  @exchange_message_handler(_MsgType.ALL_ORDERS)
  def _handle_all_orders(self, exchange_message: PrivateExchangeMessage):
    """{'status': 'error', 'err_code': 1032, 'err_msg': 'The number of access exceeded the limit.', 'ts': 1545046221833}"""
    timestamp = get_timestamp()
    response = exchange_message.data
    if response is None:
      return

    if response.status_code != requests.codes.ok:
      self._handle_error_response(response)
      return

    message = response.json()
    if message['status'] != 'ok':
      self._logger.error('status is not ok: %s' % message)
      return

    for entry in message['data']['orders']:
      self._handle_raw_order_info(entry, timestamp)

  @exchange_message_handler(_MsgType.ORDERS_INFO)
  def _handle_orders_info(self, exchange_message: PrivateExchangeMessage):
    timestamp = get_timestamp()
    response = exchange_message.data
    if response is None:
      return

    if response.status_code != requests.codes.ok:
      self._handle_error_response(response)
      return

    message = response.json()
    if message['status'] != 'ok':
      self._logger.error('status is not ok: %s' % message)
      return

    for entry in message['data']:
      self._handle_raw_order_info(entry, timestamp)

  @exchange_message_handler(_MsgType.WS_ORDER_UPDATE)
  def _handle_ws_orders_update(self, exchange_message: PrivateExchangeMessage):
    timestamp = exchange_message.received_timestamp
    data = exchange_message.data
    self._handle_raw_order_info(data, timestamp)

  def _handle_raw_order_info(self, raw_order_info, timestamp):
    native_symbol = raw_order_info['contract_code']
    product = HuobiFuturesProduct.FromStrNativeProduct(native_symbol)
    if product not in self._og_info.products:
      return  # ignore

    order_info = create_order_from_open_order_message(raw_order_info, timestamp)
    exchange_order_id = order_info.internal.exchange_order_id
    order = self._order_manager.get(exchange_order_id=exchange_order_id)
    fill_type = None
    if order is None:
      order_id = self._order_id_generator.gen_order_id(is_foreign=True)
      order_info.order_id = order_id
      self._order_manager.add_order(order_info)
      order = order_info
    else:
      role = raw_order_info.get('role')
      if role == 'maker':
        fill_type = coin_order_pb2.MAKER_FILL_TYPE
      elif role == 'taker':
        fill_type = coin_order_pb2.TAKER_FILL_TYPE
      elif role is None:
        fill_type = None
      else:
        self._logger.error('Unknown role type! %s', raw_order_info)
        fill_type = None
      order_info.order_id = order.order_id
      order.internal.fee = order_info.internal.fee
    self._update_order(order, order_info, timestamp, fill_type=fill_type)

  @exchange_message_handler(_MsgType.REST_ACCOUNT_BALANCE)
  def _handle_rest_balance(self, exchange_message: PrivateExchangeMessage):
    response = exchange_message.data
    if response.status_code != requests.codes.ok:
      self._handle_error_response(response)
      return

    message = response.json()
    if message['status'] != 'ok':
      self._logger.error('status is not ok: %s' % message)
      return

    self._og_info.update_balance(message)
    self._og_logger.write_balance(self._og_info.balance_as_proto())
    if not self.is_ready():
      self._og_ready.set_balance_ready(True)
      self._og_ready.check_ready()

  @exchange_message_handler(_MsgType.REST_ACCOUNT_POSITION)
  def _handle_rest_position(self, exchange_message: PrivateExchangeMessage):
    response = exchange_message.data
    if response.status_code != requests.codes.ok:
      self._handle_error_response(response)
      return

    message = response.json()
    if message['status'] != 'ok':
      self._logger.error('status is not ok: %s' % message)
      return

    self._og_info.update_position(message)
    self._og_logger.write_position(self._og_info.position_as_proto())
    if not self.is_ready():
      self._og_ready.set_position_ready(True)
      self._og_ready.check_ready()
